package com.zfei.firstflink;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.lang.reflect.Array;
import java.util.Arrays;

/***
 *$com.zfei.TestSet
 *@author:jiwenfei
 *
 *@version:
 *
 *createTime:2018/8/16:下午1:15
 *
 *description:
 */
public class TestSet {
    public static void main(String[] args) throws Exception {

        ExecutionEnvironment see=ExecutionEnvironment.getExecutionEnvironment();
        DataSet<Integer> input = see.fromElements(1, 2, 3);

        input.map(item->item*2).print();
        DataSet<String >ff=input.flatMap(new FlatMapFunction<Integer, String>() {
            @Override
            public void flatMap(Integer num, Collector<String> out) throws Exception {
                    StringBuilder sb=new StringBuilder();
                    for (int i = 0; i <num ; i++) {
                        sb.append("a");
                    }
                    out.collect(sb.toString());
            }
        });
        ff.print();


        DataSet<String> input2 = see.fromElements("Please count", "the words the", "but not this");
        input2.filter(t->!t.contains("not"))
                .map(item->item.split(" "))
                .flatMap(new FlatMapFunction<String[], Tuple2<String,Integer>>() {
                    @Override
                    public void flatMap(String[]wordArr,Collector<Tuple2<String, Integer>> out){
                        Arrays.stream(wordArr).forEach(t->out.collect(new Tuple2<>(t,1)));
                    }
                }).groupBy(0)
                .sum(1)
                .print();

    }
}
